/*
 *  Arrow
 *
 *  Created by David Giovannini on 1/5/07.
 *  Copyright 2007 Software by Jove. All rights reserved.
 *
 */
 
#include "SBJ/Thread/PlatformDetails.h"

#include <boost/bind.hpp>

namespace SBJ
{

template < typename Task >
TaskDispatcher< Task >::TaskDispatcher( int numThreads, bool clearUnusedOnDestroy )
: m_excObserver( new DefaultExceptionObserver )
, m_pending( 0 )
, m_clearUnusedOnDestroy( clearUnusedOnDestroy )
{
	// Create the threads
	for ( int i = 0; i < numThreads; ++i )
	{
		m_threads.push_back( 
			new boost::thread( boost::bind( &TaskDispatcher::threadWork, 
											boost::ref( *this ),
											i ) ) );
	}
}

template < typename Task >
TaskDispatcher< Task >::~TaskDispatcher( void )
{
	// Wait until running tasks are complete
	// All threads will be in wait state when done
	if ( m_clearUnusedOnDestroy == true )
	{
		clear_remaining_tasks();
	}
	join_all_tasks();

	// Wake them all up with nothing to do
	m_start.notify_all();

	// Wake up any other join-all threads 
	m_stop.notify_all();

	// Delete the threads
	for (std::vector<boost::thread*>::iterator it = m_threads.begin();
			it != m_threads.end(); ++it)
	{
		// Allow the thread to exit nicely
		(*it)->join();
		// Delete it
		delete (*it);
	}
}

template < typename Task >
bool
TaskDispatcher< Task >::invariant( void ) const
{
	return true;
}

template < typename Task >
void
TaskDispatcher< Task >::addTask( Task& task )
{
	// The special case for zero threads, execute immediately
	if ( m_threads.size() == 0 )
	{
		try
		{
			task();
		}
		catch ( std::exception& exception )
		{
			m_excObserver->onException( exception, typeid( Task ).name() );
		}
		catch ( ... )
		{
			m_excObserver->onException( typeid( Task ).name() );		
		}
	}
	else
	{
		// Enqueue the task
		boost::mutex::scoped_lock lock( m_taskGuard );
		try
		{
			m_tasks.push_back( task );
		}
		catch ( std::exception& exception )
		{
			m_excObserver->onException( exception, typeid( Task ).name() );
		}
		catch ( ... )
		{
			m_excObserver->onException( typeid( Task ).name() );		
		}
		++m_pending;
		// start up any thread
		m_start.notify_one();
	}
}

template < typename Task >
void
TaskDispatcher< Task >::addPolyTask( PolyTask* task )
{
	PolyTaskRef taskRef( task ); 
	addTask( taskRef );
}

template < typename Task >
void
TaskDispatcher< Task >::join_all_tasks( void )
{
	boost::mutex::scoped_lock lock( m_taskGuard );
	if ( m_threads.size() == 0 ) return;

	// If we have any tasks being worked or in the queue
	if ( m_pending > 0 )
	{
		m_stop.wait( lock );
	}
}

template < typename Task >
void
TaskDispatcher< Task >::clear_remaining_tasks( void )
{
	boost::mutex::scoped_lock lock( m_taskGuard );
	// "Complete" the unused tasks
	m_pending -= static_cast< int >( m_tasks.size() );
	try
	{
		m_tasks.clear();
	}
	catch ( std::exception& exception )
	{
		m_excObserver->onException( exception, typeid( Task ).name() );
	}
	catch ( ... )
	{
		m_excObserver->onException( typeid( Task ).name() );		
	}
}

template < typename Task >
void
TaskDispatcher< Task >::threadWork( int threadIndex )
{
	PlatformDetails::ThreadGuard threadGuard;
	while ( true )
	{
		Tasks tempList;
		{
			boost::mutex::scoped_lock lock( m_taskGuard );
			// If no tasks available
			if ( m_tasks.size() == 0 )
			{
				m_start.wait( lock );
			}
			// Wake up with nothing queued can only mean one thing
			if ( m_tasks.size() == 0 )
			{
				break;
			}
			// Get the task
			tempList.splice( tempList.begin(), m_tasks, m_tasks.begin() );
		}
		try
		{
			tempList.front()();
		}
		catch ( std::exception& exception )
		{
			m_excObserver->onException( exception, typeid( Task ).name() );
		}
		catch ( ... )
		{
			m_excObserver->onException( typeid( Task ).name() );		
		}
		// Record completion
		{
			boost::mutex::scoped_lock lock( m_taskGuard );
			--m_pending;
			// Wake up the join-all threads if last one is done
			if ( m_pending == 0 )
			{
				m_stop.notify_all();
			}
		}
	}
}

} // SBJ
